package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.constant.Constant;

import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.utill.DateFormatUtil;
import com.atguigu.gmall.realtime.common.utill.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(10022, 4, "dws_traffic_vc_ch_ar_is_new_page_view_window", Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }


    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 1.对流中数据进行类型转换    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        //jsonObjDS.print();

        //TODO 2.按照mid进行分组
        KeyedStream<JSONObject, String> midKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //TODO 3.jsonObj->实体类对象   使用Flink状态编程判断是否为独立访客  相当于WordCount的封装二元组
        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = midKeyedDS.process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
            private ValueState<String> lastVisitDateState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("lastVisitDateState", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                String vc = commonJsonObj.getString("vc");
                String ch = commonJsonObj.getString("ch");
                String ar = commonJsonObj.getString("ar");
                String isNew = commonJsonObj.getString("is_new");

                JSONObject pageJsonObj = jsonObj.getJSONObject("page");

                //判断是否为当天独立访客
                String lastVisitDate = lastVisitDateState.value();
                Long ts = jsonObj.getLong("ts");
                String curVisitdate = DateFormatUtil.tsToDate(ts);
                Long uvCt = 0L;
                if (StringUtils.isEmpty(lastVisitDate) || !lastVisitDate.equals(curVisitdate)) {
                    uvCt = 1L;
                    lastVisitDateState.update(curVisitdate);
                }
                //会话计数
                String lastPageId = pageJsonObj.getString("last_page_id");
                Long svCt = StringUtils.isEmpty(lastPageId) ? 1L : 0L;

                TrafficPageViewBean viewBean = new TrafficPageViewBean("", "", "", vc, ch, ar, isNew, uvCt, svCt, 1L, pageJsonObj.getLong("during_time"), ts);
                out.collect(viewBean);
            }
        });
        //beanDS.print();

        //TODO 4.指定Watermark生成策略以及提取事件时间字段

        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forMonotonousTimestamps()
                //.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean bean, long recordTimestamp) {
                        return bean.getTs();
                    }
                })

        );

        //TODO 5.按照统计的维度进行分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> dimKeyedDS = withWatermarkDS.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean bean) throws Exception {
                return Tuple4.of(bean.getVc(), bean.getCh(), bean.getAr(), bean.getIsNew());
            }
        });

        //TODO 6.开窗-------------------??????
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS = dimKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 7.聚合----------------------??????????
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                return value1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                TrafficPageViewBean viewBean = input.iterator().next();
                String stt = DateFormatUtil.tsToDateTime(window.getStart());
                String edt = DateFormatUtil.tsToDateTime(window.getEnd());
                String curDate = DateFormatUtil.tsToDate(window.getStart());
                viewBean.setStt(stt);
                viewBean.setEdt(edt);
                viewBean.setCur_date(curDate);
                out.collect(viewBean);
            }
        });

        //TODO 8.将聚合结果写到Doris中
        reduceDS.print();
        reduceDS.map(new BeanToJsonStrMapFunction<>())                //将流中实体类对象转换为json格式字符串
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_ch_ar_is_new_page_view_window"));

    }
}
